/*
 * Su-yun Information Technology, Co.,Ltd CONFIDENTIAL
 *
 * (C) Copyright Su-yun Information Technology, Co.,Ltd. 2021 All Rights Reserved.
 *
 * NOTICE:  All information contained herein is, and remains the property of
 * Su-yun Information Technology, Co.,Ltd. The intellectual property herein
 * are protected by trade secret, patent law or copyright law. Dissemination
 * of this information or reproduction of this material is strictly forbidden
 * unless prior written permission is obtained from Su-yun Information Technology.
 *
 * https://suyunplat.com/
 *
 */
package com.xtiim.battery.discharge.common.concurrent.pool;

import lombok.extern.slf4j.Slf4j;

import java.util.Collection;
import java.util.List;
import java.util.concurrent.*;

/**
 * High performance thread pool implementation.
 *
 * @param <T>
 */
@Slf4j
public class LtqExecutor<T> implements ExecutorService {

    private static final Integer DEFAULT_MAX_QUEUE_SIZE = 5000;
    private volatile boolean running;
    private volatile boolean stopped;
    private final int threadCount;
    private final BlockingQueue<Runnable> queue;
    private final Thread[] threads;
    private final int maxQueueSize;

    public LtqExecutor(int threadCount) {
    	this(threadCount, Executors.defaultThreadFactory(), DEFAULT_MAX_QUEUE_SIZE);
    }

    public LtqExecutor(int threadCount, ThreadFactory threadFactory, int maxQueueSize) {
        this.threadCount = threadCount;
        this.queue = new LinkedTransferQueue<>();
        this.maxQueueSize = maxQueueSize;
        running = true;
        stopped = false;
        threads = new Thread[threadCount];
        for(int i = 0; i < threadCount; i++) {
            threads[i] = threadFactory.newThread(new Worker());
            threads[i].start();
        }
    }

    @Override
    public void execute(Runnable runnable) {
        try {
            if (queue.size() >= maxQueueSize) {
                log.error("Task queue size exceeds limit " + maxQueueSize + ", discard this task.");
                return;
            }
            queue.put(runnable);
	    } catch (InterruptedException ie) {
	    	throw new RuntimeException(ie);
	    }
        log.info("ltq exec tName:{}, qSize:{}", Thread.currentThread().getName(), queue.size());
    }

    private class Worker implements Runnable {
        @Override
        public void run() {
            while(running) {
                Runnable runnable = null;
                try {
                    runnable = queue.take();
                } catch (InterruptedException ie) {
                    // was interrupted - just go round the loop again,
                    // if running = false will cause it to exit
                }
                try {
                    if (runnable != null) {
                    	runnable.run();
                    }
                } catch (Exception e) {
                    // just continue. theoretically, exception should be catch within runnable itself.
                }
                log.info("ltq take tName:{}, qSize:{}", Thread.currentThread().getName(), queue.size());
            }
        }
    }

    @Override
    public void shutdown() {
        running = false;
        for(int i = 0; i < threadCount; i++) {
            threads[i].interrupt();
            threads[i] = null;
        }
        stopped = true;
    }

    @Override
    public boolean isShutdown() {
        return running;
    }

    @Override
    public boolean isTerminated() {
        return stopped;
    }

    @Override
    public Future submit(Callable task) {
    	if (task == null) {
    		throw new NullPointerException("The task cloudn't be null");
    	}
        RunnableFuture ftask = newTask(task);
        execute(ftask);
        return ftask;
    }

    @Override
    public Future submit(Runnable task) {
    	if (task == null) {
    		throw new NullPointerException("The task cloudn't be null");
    	}
        RunnableFuture<Void> ftask = newTask(task, null);
        execute(ftask);
        return ftask;
    }

    protected <T> RunnableFuture<T> newTask(Runnable runnable, T value) {
        return new FutureTask<T>(runnable, value);
    }

    protected <T> RunnableFuture<T> newTask(Callable<T> callable) {
        return new FutureTask<T>(callable);
    }

    @Override
    public boolean awaitTermination(long timeout, TimeUnit unit) {
        throw new UnsupportedOperationException("oops!");
    }

    @Override
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) {
        throw new UnsupportedOperationException("oops!");
    }

    @Override
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) {
        throw new UnsupportedOperationException("oops!");
    }

    @Override
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks) {
        throw new UnsupportedOperationException("oops!");
    }

    @Override
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) {
        throw new UnsupportedOperationException("oops!");
    }

    @Override
    public List<Runnable> shutdownNow() {
        throw new UnsupportedOperationException("oops!");
    }

    @Override
    public <T> Future<T> submit(Runnable task, T result) {
        throw new UnsupportedOperationException("oops!");
    }

    @Override
    public String toString() {
        throw new UnsupportedOperationException("oops!");
    }
}
